

#include <stdio.h>
#include <string.h>
#include <string>
#include <list>
#include <gio/gio.h>
#include <glib.h>
#include <netinet/in.h>
#include <unistd.h>

#include "xm_log.h"
#include "xm_crash.h"
#include "xm_version.h"
#include "xm_app_config.h"
#include "xdtls_srtp.h"


//
// rtc signal server
//

#include <random>
#include <libsoup/soup.h>
#include "xrtc_endpoint.h"
#include "xcutil.h"
#include "rapidjson/document.h"
#include "rapidjson/writer.h"
#include "rapidjson/stringbuffer.h"
using namespace rapidjson;


#define mgr_dbgv(FMT, ARGS...) dbgv("|cmgr| " FMT, ##ARGS )
#define mgr_dbgi(FMT, ARGS...) dbgi("|cmgr| " FMT, ##ARGS )
#define mgr_dbge(FMT, ARGS...) dbge("|cmgr| " FMT, ##ARGS )

#define ZERO_ALLOC_(o, type, sz) do{o=(type)malloc(sz); memset(o, 0, sz);}while(0)
#define MAX_OBJ_ID_LEN 32



# pragma mark - confr_mgr declaration

class XConference;
class XSession;
class XRtcc;
class XStream;
typedef std::map<std::string, XConference *> XConferenceMap;
typedef std::map<std::string, XSession *> XSessionMap;
typedef std::map<std::string, XRtcc *> XRtccMap;
typedef std::map<std::string, XStream *> XStreamMap;


typedef struct confr_manager_st * confr_manager_t;
typedef void (* confr_mgr_callback_on_send)(confr_manager_t mgr, void * main_ctx, void * conn_ctx, const char * msg, int msg_len);
typedef struct confr_mgr_callbacks{
    void * main_ctx;
    confr_mgr_callback_on_send on_send;
}confr_mgr_callbacks;

struct confr_manager_st{
    confr_mgr_callbacks callbacks;
    XConferenceMap * confrs;
    XSessionMap * sessions;
    XSessionMap * conn_sessions;
};



# pragma mark - utils

static inline
uint64_t get_next_obj_id(){
    static uint64_t next_id = 0;
    return ++next_id;
}

static inline
char generate_random_char(){
    static std::random_device rd;
    return 'a' + rd()%26;
}
static inline
int generate_obj_id(const char * prefix, char * buf, size_t bufsz){
    return snprintf(buf, bufsz, "%s-%lld-%c%c%c"
                    , prefix?prefix:"obj"
                    , get_next_obj_id()
                    , generate_random_char()
                    , generate_random_char()
                    , generate_random_char()
                    );
}

static inline
const char * generate_next_id(const char * prefix, uint64_t& next_id){
    static char buf[64];
    static int bufsz = sizeof(buf);
    ++next_id;
    snprintf(buf, bufsz, "%s-%lld-%c%c%c"
             , prefix
             , next_id
             , generate_random_char()
             , generate_random_char()
             , generate_random_char()
             );
    return buf;
}

static inline
const char * generate_tsx_id(){
    static uint64_t next_id = 0;
    return generate_next_id("xtsx", next_id);
}

static inline
const char * generate_session_id(){
    static uint64_t next_id = 0;
    return generate_next_id("xsess", next_id);
}

static inline
const char * generate_member_id(){
    static uint64_t next_id = 0;
    return generate_next_id("xmem", next_id);
}

static inline
const char * generate_stream_id(){
    static uint64_t next_id = 0;
    return generate_next_id("xstream", next_id);
}

template<class keyT, class ValueT>
ValueT map_get(std::map<keyT, ValueT> * m, keyT uid, bool is_remove, ValueT nullvalue){
    auto it = m->find(uid);
    if(it == m->end()){
        return nullvalue;
    }
    ValueT v = it->second;
    if(is_remove){
        m->erase(it);
    }
    return v;
}


template<class ValueT>
ValueT* map_get_p_by_str(std::map<std::string, ValueT*> * m, const std::string& uid, bool is_remove){
    return map_get<std::string, ValueT*>(m, uid, is_remove, NULL);
}

// utils <==


# pragma mark - OPPair

enum XSIG_REASON_TYPE{
    XSIG_REASON_NORMAL = 0,
    XSIG_REASON_NORESPONSE = 1,
    XSIG_REASON_REJECT = 2,
    XSIG_REASON_BUSY = 3,
    XSIG_REASON_FAIL = 4,
    XSIG_REASON_UNSUPPORTED = 5,
    XSIG_REASON_OTHER_DEVICE = 100,
    XSIG_REASON_DISMISS = 101, // conference close
    XSIG_REASON_TIMEOUT = 102,
    XSIG_REASON_MAX = 10000
};


enum xsignal_op_t{
    XSIG_OP_ENTER = 200,
    XSIG_OP_PINGC = 100,
    XSIG_OP_INITC = 102,
    XSIG_OP_ACPTC = 104,
    XSIG_OP_TCKLC = 105,
    XSIG_OP_ANSC = 106,
    XSIG_OP_TERMC = 107,
    XSIG_OP_SUBC = 205,
    XSIG_OP_USUBC = 206,
    XSIG_OP_GETC = 1000,
    XSIG_OP_EXIT = 201,
    XSIG_OP_RSP = 1001,
    XSIG_OP_RCTRLC = 1002,
    XSIG_OP_INTERESTC = 1003,
    
    XSIG_ON_ENTER = 300,
    XSIG_ON_PUB = 302,
    XSIG_ON_UNPUB = 303,
//    XSIG_ON_MEMBERS = 304,
    XSIG_ON_EXIT = 301,
//    XSIG_ON_CLOSE = 204, // conference close
    
    XSIG_OP_STREAM_CONTROL = 400,
//    XSIG_ON_JOIN = (401) //

    
    XSIG_OP_END
};



class OPPair{
public:
    typedef std::map<std::string, const OPPair *> OPMap;
    typedef std::map<int, const OPPair *> OPNMap;
    
    static inline const OPPair** getAll();
    static inline const OPPair * get(int opnum);
    static inline const OPPair * get(const std::string& opname);
    
    OPPair(const std::string& opname, int opnum, bool ismute=false)
    : opname_(opname), opnum_(opnum), ismute_(ismute)
    { }
    const std::string opname_;
    const int opnum_;
    const bool ismute_;
};


static OPPair * ENTER_OP = new OPPair("enter", XSIG_OP_ENTER);
static OPPair * PINGC_OP = new OPPair("pingC", XSIG_OP_PINGC, true);
static OPPair * INITC_OP = new OPPair("initC", XSIG_OP_INITC);
static OPPair * ACEPT_OP = new OPPair("acptC", XSIG_OP_ACPTC);
static OPPair * TCKLC_OP = new OPPair("tcklC", XSIG_OP_TCKLC);
static OPPair * ANSC_OP = new OPPair("ansC", XSIG_OP_ANSC);
static OPPair * TERMC_OP = new OPPair("termC", XSIG_OP_TERMC);
static OPPair * SUBC_OP = new OPPair("subC", XSIG_OP_SUBC);
static OPPair * UNSUBC_OP = new OPPair("usubC", XSIG_OP_USUBC);
static OPPair * GETC_OP = new OPPair("getC", XSIG_OP_GETC, false);
static OPPair * EXIT_OP =  new OPPair("exit", XSIG_OP_EXIT);
static OPPair * RSP_OP = new OPPair("rsp", XSIG_OP_RSP);
static OPPair * STREAMCTRL_OP = new OPPair("streamC", XSIG_OP_STREAM_CONTROL);
static OPPair * REMOTECTRL_OP = new OPPair("remoteC", XSIG_OP_RCTRLC);
static OPPair * INTEREST_OP = new OPPair("interestC", XSIG_OP_INTERESTC);

static OPPair * ONENTER_OP = new OPPair("onenter", XSIG_ON_ENTER);
static OPPair * ONPUB_OP =  new OPPair("onpub", XSIG_ON_PUB);
static OPPair * ONUNPUB_OP =  new OPPair("onupub", XSIG_ON_UNPUB);
static OPPair * ONEXIT_OP = new OPPair("onexit", XSIG_ON_EXIT);


//static OPPair * ONCLOSE_OP = new OPPair("onclose", XSIG_ON_CLOSE); // conference close
// new OPPair("onmems", XSIG_ON_MEMBERS)
// new OPPair("onctrl", XSIG_ON_CONTROL)

const OPPair** OPPair::getAll(){
    static const OPPair * pairs[] = {
        ENTER_OP
        , PINGC_OP
        , INITC_OP
        , ACEPT_OP
        , TCKLC_OP
        , ANSC_OP
        , TERMC_OP
        , SUBC_OP
        , UNSUBC_OP
        , GETC_OP
        , EXIT_OP
        , RSP_OP
        , STREAMCTRL_OP
        , REMOTECTRL_OP
        , INTEREST_OP
        
        , ONENTER_OP
        , ONPUB_OP
        , ONUNPUB_OP
        , ONEXIT_OP
//        , ONCLOSE_OP

        , NULL
    };
    return pairs;
}

const OPPair * OPPair::get(const std::string& opname){
    static OPPair::OPMap * m = NULL;
    if(!m){
        m = new OPPair::OPMap;
        const OPPair** pairs = OPPair::getAll();
        for(int i = 0; pairs[i]; i++){
            (*m)[pairs[i]->opname_] = pairs[i];
        }
    }
    return map_get_p_by_str<const OPPair>(m, opname, false);
}

const OPPair * OPPair::get(int opnum){
    static OPPair::OPNMap * m = NULL;
    if(!m){
        m = new OPPair::OPNMap;
        const OPPair** pairs = OPPair::getAll();
        for(int i = 0; pairs[i]; i++){
            (*m)[pairs[i]->opnum_] = pairs[i];
        }
    }
    return map_get<int, const OPPair*>(m, opnum, false, NULL);
}

// OPPair <==



# pragma mark - entitis definition
class XObject{
public:
    XObject(const std::string& uidPrefix){
        char obj_id[MAX_OBJ_ID_LEN];
        generate_obj_id(uidPrefix.c_str(), obj_id, sizeof(obj_id));
        this->uid_ = obj_id;
    }
    virtual ~XObject(){}
    std::string uid_;
};

class XStream: public XObject{
public:
    XStream( XRtcc * rtcc)
    : XObject("stream")
    , rtcc_(rtcc)
    , type_(0), voff_(0), aoff_(0){ uid_ = generate_stream_id();}
    
public:
    XRtcc * rtcc_;
    
    int type_;
    int voff_;
    int aoff_;
    
    std::string name_;
    XRtccMap subRtccs_;  // subscriber
};

class XRtcc: public XObject{
public:
    XRtcc(XSession * session, const std::string& uidPrefix)
    : XObject(uidPrefix)
    , session_(session)
    , pubStream_(NULL)
    , subStream_(NULL)
    , rtc_(NULL)
    { }
    
public:
    XSession * session_;
    XStream * pubStream_;
    XStream * subStream_;
    rtc_endpoint_t rtc_;
    std::string remoteRtcId_;
    std::string subStreamId_;
};


class XSession: public XObject{
public:
    XSession(XConference * confr)
    : XObject("sess")
    , confr_(confr)
    , isJsonObjMode_(false)
    , connctx_(NULL)
    , isCheckLive_(false)
    , lastCheckLiveTime(get_timestamp_ms())
    , isEntered(false)
    { uid_ = generate_session_id(); }
public:
    XConference * confr_;
    bool isJsonObjMode_;
    
    std::string  memberId_;
    std::string  memberName_;
    XRtccMap rtccs_;
    
    std::string connId_;
    void * connctx_;
    bool isCheckLive_;
    int64_t lastCheckLiveTime;
    bool isEntered;
    
};


class XConference: public XObject{
public:
    XConference(const std::string& confrId) : XObject("confr"), cver_(0)
    , isCheckLive_(false)
    , lastCheckLiveTime(get_timestamp_ms())
    {
        uid_ = confrId;
    }
    
public:
    XSessionMap sessions_;
    XStreamMap streams_;
    XSessionMap members_;  // visible, by memberName
    XSessionMap memberIds_;  // by memberId
    uint64_t cver_;
    bool isCheckLive_;
    int64_t lastCheckLiveTime;
};

// entitis definition <==


# pragma mark - proto definition

#define MAKE_TYPE_FLAG(t) (1 << (t))
#define MK_ITER_CSTR(iter) std::string((iter)->value.GetString(), (iter)->value.GetStringLength())
#define MK_ITER_INT(iter) (iter)->value.GetInt()

static const char * proto_type_names[] = {
    "kNullType",
    "kFalseType",
    "kTrueType",
    "kObjectType",
    "kArrayType",
    "kStringType",
    "kNumberType"
};

class XProtoField{
public:
    XProtoField(const std::string& name, bool mandatory, uint32_t typflag)
    : name_(name), mandatory_(mandatory), typflag_(typflag)
    { }
    const std::string name_;
    bool mandatory_;
    const uint32_t typflag_;
    Value::ConstMemberIterator iter_;
};


class XProtoBase{
public:
    XProtoBase(const std::string& name):name_(name){}
    virtual ~XProtoBase(){}
    virtual std::vector<XProtoField*>* getFields(){
        return &fields_;
    }
    virtual const std::string * getName(){
        return &name_;
    }
    int parse(const Value &doc_, std::ostringstream& oss_err){
        XProtoBase * proto = this;
        std::vector<XProtoField*>* fields = proto->getFields();
        for(auto f : *fields){
            f->iter_  = doc_.FindMember(f->name_.c_str());
            if(f->iter_ != doc_.MemberEnd() && f->iter_->value.GetType() == kNullType){
                f->iter_ = doc_.MemberEnd();
            }
            
            if(f->iter_ == doc_.MemberEnd()){
                if(f->mandatory_){
                    oss_err << "non-exist field [" << name_ << "]->["<< f->name_ << "]";
                    return -1;
                }
                continue;
            }
            
            if(!(MAKE_TYPE_FLAG(f->iter_->value.GetType()) & f->typflag_)){
                oss_err << "unexpectd field type [" << name_ << "]->["<< f->name_ << "]: ["
                    << proto_type_names[f->iter_->value.GetType()] << "]";
                return -1;
            }
        }
        return 0;
    }
    int parse(const char * json, int json_len, std::ostringstream& oss_err){
        doc_.Parse(json);
        if(!doc_.IsObject()){
            oss_err << "json is NOT object [" << name_ << "]";
            return -1;
        }
        return parse(doc_, oss_err);
    }
    int parse(XProtoField& field, std::ostringstream& oss_err){
        const Value& strval = field.iter_->value;
        if(strval.IsString()){
            return parse(strval.GetString(), strval.GetStringLength(), oss_err);
        }else if(strval.IsObject()){
            return parse(strval, oss_err);
        }else{
            oss_err << "NOT string and NOT object [" << name_ << "]->["<< field.name_ << "]";
            return -1;
        }
    }
    
protected:
    std::string name_;
    std::vector<XProtoField*>fields_;
    Document doc_;
};

class XProtoInit : public XProtoBase{
public:
    XProtoInit(const char * name)
    : XProtoBase(name)
    , tkt_("tkt", false, MAKE_TYPE_FLAG(kStringType) | MAKE_TYPE_FLAG(kObjectType))
    , sessId_("sessId", false, MAKE_TYPE_FLAG(kStringType) )
    , rtcId_("rtcId", false, MAKE_TYPE_FLAG(kStringType) )
    , sdp_("sdp", false, MAKE_TYPE_FLAG(kStringType) | MAKE_TYPE_FLAG(kObjectType))
    , cands_("cands", false, MAKE_TYPE_FLAG(kArrayType) )
    , pubS_("pubS", false, MAKE_TYPE_FLAG(kObjectType) )
    , subSId_("subSId", false, MAKE_TYPE_FLAG(kStringType) )
    
    {
        fields_.push_back(&tkt_);
        fields_.push_back(&sessId_);
        fields_.push_back(&rtcId_);
        fields_.push_back(&sdp_);
        fields_.push_back(&cands_);
        fields_.push_back(&pubS_);
        fields_.push_back(&subSId_);
    }

    XProtoField tkt_;
    XProtoField sessId_;
    XProtoField rtcId_;
    XProtoField sdp_;
    XProtoField cands_;
    XProtoField pubS_;
    XProtoField subSId_;
};


class XProtoEnter : public XProtoInit{
public:
    XProtoEnter(): XProtoInit("enter")
    {
        tkt_.mandatory_ = true;
    }

};

class XProtoInitC : public XProtoInit{
public:
    XProtoInitC(): XProtoInit("initC")
    {
    }
    
};


class XProtoTicket : public XProtoBase{
public:
    XProtoTicket(const char * name = NULL)
    : XProtoBase(name?name:"tkt")
    , url_("url", true, MAKE_TYPE_FLAG(kStringType) )
    , type_("type", true, MAKE_TYPE_FLAG(kStringType) )
    , memName_("memName", true, MAKE_TYPE_FLAG(kStringType) )
    , confrId_("confrId", true, MAKE_TYPE_FLAG(kStringType) )
    {
        fields_.push_back(&url_);
        fields_.push_back(&type_);
        fields_.push_back(&memName_);
        fields_.push_back(&confrId_);
    }
    XProtoField url_;
    XProtoField type_;
    XProtoField memName_;
    XProtoField confrId_;
};

class XProtoSDP : public XProtoBase{
public:
    XProtoSDP()
    : XProtoBase("tkt")
    , desc_("sdp", true, MAKE_TYPE_FLAG(kStringType) )
    , type_("type", true, MAKE_TYPE_FLAG(kStringType) )
    {
        fields_.push_back(&desc_);
        fields_.push_back(&type_);
    }
    XProtoField desc_;
    XProtoField type_;

};


// {type, name, voff, aoff}
class XProtoPubS : public XProtoBase{
public:
    XProtoPubS()
    : XProtoBase("pubS")
    , type_("type", false, MAKE_TYPE_FLAG(kNumberType) )
    , name_("name", false, MAKE_TYPE_FLAG(kStringType) )
    , voff_("voff", false, MAKE_TYPE_FLAG(kNumberType) )
    , aoff_("aoff", false, MAKE_TYPE_FLAG(kNumberType) )
    {
        fields_.push_back(&type_);
        fields_.push_back(&name_);
        fields_.push_back(&voff_);
        fields_.push_back(&aoff_);
    }
    XProtoField type_;
    XProtoField name_;
    XProtoField voff_;
    XProtoField aoff_;
};

class XProtoTcklC : public XProtoBase{
public:
    XProtoTcklC()
    : XProtoBase("tcklC")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , rtcId_("rtcId", true, MAKE_TYPE_FLAG(kStringType) )
    , cands_("cands", true, MAKE_TYPE_FLAG(kArrayType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&rtcId_);
        fields_.push_back(&cands_);
        
    }
    
    XProtoField sessId_;
    XProtoField rtcId_;
    XProtoField cands_;
};

class XProtoTermC : public XProtoBase{
public:
    XProtoTermC()
    : XProtoBase("termC")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , rtcId_("rtcId", true, MAKE_TYPE_FLAG(kStringType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&rtcId_);
    }
    
    XProtoField sessId_;
    XProtoField rtcId_;
};

class XProtoExit : public XProtoBase{
public:
    XProtoExit()
    : XProtoBase("exit")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , reason_("reason", false, MAKE_TYPE_FLAG(kNumberType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&reason_);
    }
    XProtoField sessId_;
    XProtoField reason_;
};

class XProtoSubC : public XProtoBase{
public:
    XProtoSubC()
    : XProtoBase("subC")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , rtcId_("rtcId", true, MAKE_TYPE_FLAG(kStringType) )
    , subSId_("subSId", true, MAKE_TYPE_FLAG(kStringType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&rtcId_);
        fields_.push_back(&subSId_);
    }
    
    XProtoField sessId_;
    XProtoField rtcId_;
    XProtoField subSId_;
};

class XProtoUnsubC : public XProtoSubC{
};

class XProtoGetC : public XProtoBase{
public:
    XProtoGetC()
    : XProtoBase("subC")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , cver_("cver", true, MAKE_TYPE_FLAG(kNumberType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&cver_);
    }
    
    XProtoField sessId_;
    XProtoField cver_;
};

class XProtoStreamCtrl : public XProtoBase{
public:
    XProtoStreamCtrl()
    : XProtoBase("streamctrl")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , rtcId_("rtcId", true, MAKE_TYPE_FLAG(kStringType) )
    , voff_("voff", false, MAKE_TYPE_FLAG(kNumberType) )
    , aoff_("aoff", false, MAKE_TYPE_FLAG(kNumberType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&rtcId_);
        fields_.push_back(&voff_);
        fields_.push_back(&aoff_);
    }
    XProtoField sessId_;
    XProtoField rtcId_;
    XProtoField voff_;
    XProtoField aoff_;
};

class XProtoInterest : public XProtoBase{
public:
    XProtoInterest(const char * name = NULL)
    : XProtoBase(name?name:"interest")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , streamId_("streamId", true, MAKE_TYPE_FLAG(kStringType) )
    , x_("x", true, MAKE_TYPE_FLAG(kNumberType) )
    , y_("y", true, MAKE_TYPE_FLAG(kNumberType) )
    , focus_("focus", false, MAKE_TYPE_FLAG(kNumberType) )
    , expo_("expo", false, MAKE_TYPE_FLAG(kNumberType) )
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&streamId_);
        fields_.push_back(&x_);
        fields_.push_back(&y_);
        fields_.push_back(&focus_);
        fields_.push_back(&expo_);
    }
    XProtoField sessId_;
    XProtoField streamId_;
    XProtoField x_;
    XProtoField y_;
    XProtoField focus_;
    XProtoField expo_;
};

class XProtoRemoteCtrl : public XProtoBase{
public:
    XProtoRemoteCtrl()
    : XProtoBase("remoteCtrl")
    , sessId_("sessId", true, MAKE_TYPE_FLAG(kStringType) )
    , memId_("memId", true, MAKE_TYPE_FLAG(kStringType))
    , arg_("arg", true, MAKE_TYPE_FLAG(kStringType))
    , tsxId_("tsxId", true, MAKE_TYPE_FLAG(kStringType))
    {
        fields_.push_back(&sessId_);
        fields_.push_back(&memId_);
        fields_.push_back(&arg_);
        fields_.push_back(&tsxId_);
    }
    XProtoField sessId_;
    XProtoField memId_;
    XProtoField arg_;
    XProtoField tsxId_;
};


// <== proto definition

# pragma mark - entitis base operation

static inline
XConference * confr_mgr_new_confr(confr_manager_t mgr, const std::string& confrId){
    mgr_dbgi("new confr [%s]", confrId.c_str());
    XConference * confr = new XConference(confrId);
    (*mgr->confrs)[confr->uid_] = confr;
    return confr;
}

static inline
XConference * confr_mgr_get_confr(confr_manager_t mgr, const std::string& confrId, bool is_remove){
    XConference * confr = map_get_p_by_str<XConference>(mgr->confrs, confrId, is_remove);
    return confr;
}

static inline
XSession * confr_mgr_new_session(confr_manager_t mgr, XConference * confr){
    XSession * session = new XSession(confr);
    mgr_dbgi("new session [%s]->[%s]", confr->uid_.c_str(), session->uid_.c_str());
    (*mgr->sessions)[session->uid_] = session;
    confr->sessions_[session->uid_] = session;
    confr->cver_++;
    return session;
}

static inline
XSession * confr_mgr_get_session(confr_manager_t mgr, const std::string& sessionId, bool is_remove){
    XSession * session = map_get_p_by_str<XSession>(mgr->sessions, sessionId, is_remove);
    if(is_remove && session){
        map_get_p_by_str<XSession>(&session->confr_->memberIds_, session->memberId_, true);
        XSession * memberSession = map_get_p_by_str<XSession>(&session->confr_->members_, session->memberName_, false);
        if(memberSession == session){
            map_get_p_by_str<XSession>(&session->confr_->members_, session->memberName_, true);
        }
        
        session->confr_->cver_++;
        map_get_p_by_str<XSession>(&session->confr_->sessions_, sessionId, is_remove);
        if(session->connId_.length() > 0){
            map_get_p_by_str<XSession>(mgr->conn_sessions, session->connId_, true);
        }
        
    }
    return session;
}

static inline
XSession * confr_mgr_get_session_by_member_id(confr_manager_t mgr, XConference * confr, const std::string& memberId, bool is_remove){
    XSession * session = map_get_p_by_str<XSession>(&confr->memberIds_, memberId, is_remove);
    if(is_remove && session){
        confr_mgr_get_session(mgr, session->uid_, is_remove);
    }
    return session;
}

static inline
XSession * confr_mgr_add_member(confr_manager_t mgr, XSession * newSession){
    XConference * confr = newSession->confr_;
    std::string& memberName = newSession->memberName_;
    XSession * oldSession = map_get_p_by_str<XSession>(&confr->members_, memberName, true);
    if(oldSession){
        confr_mgr_get_session(mgr, oldSession->uid_, true);
    }
    confr->members_[memberName] = newSession;
    confr->memberIds_[newSession->memberId_] = newSession;
    return oldSession;
}

static inline
XSession * confr_mgr_bind_session_conn(confr_manager_t mgr, XSession * session, const std::string& connId, void * conn_ctx){
    XSession * old_session = NULL;
    
    if(session->connId_ == connId && session->connctx_ == conn_ctx){
        // do nothing
        return session;
    }
    
    // first all, remove old conn session
    if(session->connId_.length() > 0){
        old_session = map_get_p_by_str<XSession>(mgr->conn_sessions, session->connId_, true);
    }
    
    if(conn_ctx){
        if(old_session){
            if(session->connId_ != connId){
                mgr_dbgi("bind conn: replace connId, [%s]->[%s][%p]",  session->connId_.c_str(), connId.c_str(), conn_ctx);
            }else{
                mgr_dbgi("bind conn: replace ctx [%p]->[%p]", session->connctx_, conn_ctx);
            }
        }else{
            mgr_dbgi("bind conn: new connId, [%s][%p]", connId.c_str(), conn_ctx);
        }
        session->connId_ = connId;
        session->connctx_ = conn_ctx;
        (*mgr->conn_sessions)[session->connId_] = session;
    }else{
        if(session->connctx_){
            mgr_dbgi("bind conn: remove ctx [%s][%p]", session->connId_.c_str(), session->connctx_);
        }
        
        session->connId_ = "";
        session->connctx_ = NULL;
    }
    
    return old_session;
}

static inline
XSession * confr_mgr_get_session_conn(confr_manager_t mgr, const std::string& connId){
    return map_get_p_by_str<XSession>(mgr->conn_sessions, connId, false);
}

static inline
XStream * confr_mgr_new_stream(confr_manager_t mgr, XRtcc* rtcc){
    std::string uidPrefix;
    XStream * stream = new XStream(rtcc);
    rtcc->pubStream_ = stream;
    rtcc->session_->confr_->streams_[stream->uid_] = stream;
    stream->rtcc_->session_->confr_->cver_++;
    mgr_dbgi("new stream [%s]", stream->uid_.c_str());
    return stream;
}

static inline
XStream * confr_mgr_get_stream(confr_manager_t mgr, XConference* confr, const std::string& streamId, bool is_remove){
    XStream * stream = map_get_p_by_str<XStream>(&confr->streams_, streamId, is_remove);
    if(is_remove && stream){
        stream->rtcc_->session_->confr_->cver_++;
        stream->rtcc_->pubStream_ = NULL;
        for(auto o : stream->subRtccs_){
            XRtcc * rtcc = o.second;
            rtcc->subStream_ = NULL;
        }
    }
    return stream;
}

static inline
std::string confr_mgr_make_sub_rtcc_uid(XRtcc * rtcc){
    return rtcc->session_->uid_ + "-" + rtcc->uid_;
}

static inline
XStream * confr_mgr_sub_stream(confr_manager_t mgr, XStream * stream, XRtcc * rtcc){
    std::string subUid = confr_mgr_make_sub_rtcc_uid(rtcc);
    XRtcc * r = map_get_p_by_str<XRtcc>(&stream->subRtccs_,subUid, false);
    if(r){
        // already sub
        return NULL;
    }
    rtcc->subStream_ = stream;
    rtcc->subStreamId_ = stream->uid_;
    stream->subRtccs_[subUid] = rtcc;
    return stream;
}

static inline
XStream * confr_mgr_unsub_stream(confr_manager_t mgr, XStream * stream, XRtcc * rtcc){
    std::string subUid = confr_mgr_make_sub_rtcc_uid(rtcc);
    XRtcc * r = map_get_p_by_str<XRtcc>(&stream->subRtccs_, subUid, true);
    if(!r){
        // NOT sub
        return NULL;
    }
    rtcc->subStream_ = NULL;
    rtcc->subStreamId_ = "";
    return stream;
}


static inline
std::string confr_mgr_make_rtcc_uid(const std::string& remoteRtcId){
    return remoteRtcId;
}

static inline
XRtcc * confr_mgr_new_rtcc(confr_manager_t mgr, XSession * session, const std::string& remoteRtcId){
    std::string uidPrefix;
    XRtcc * rtcc = new XRtcc(session, remoteRtcId);
    rtcc->uid_ = confr_mgr_make_rtcc_uid(remoteRtcId);
    session->rtccs_[rtcc->uid_] = rtcc;
    mgr_dbgi("new rtcc [%s]", rtcc->uid_.c_str());
    return rtcc;
}
    
static inline
XRtcc * confr_mgr_get_rtcc(confr_manager_t mgr, XSession * session, const std::string& rtccId, bool is_remove){
    XRtcc * rtcc = map_get_p_by_str<XRtcc>(&session->rtccs_, rtccId, is_remove);
    if(rtcc && is_remove){
        if(rtcc->subStream_){
            confr_mgr_unsub_stream(mgr, rtcc->subStream_, rtcc);
        }
    }
    return rtcc;
}

// <== entitis base operation


# pragma mark - broadcast events

static inline
int confr_mgr_send_msg(confr_manager_t mgr, void * conn_ctx, const char * msg, int msg_len){
    if(!mgr->callbacks.on_send) return 0;
    mgr_dbgi("signal msg: notify ==> %.*s", msg_len, msg);
    mgr->callbacks.on_send(mgr, mgr->callbacks.main_ctx, conn_ctx, msg, msg_len);
    return 0;
}


static inline
int confr_mgr_broadcast_to_sessions(confr_manager_t mgr, XSession * session, const OPPair * op, char * msg_buf, int base_len, bool sendtoSelf=false){
    if(!mgr->callbacks.on_send) return 0;
    
    mgr_dbgi("broadcast ev [%s]->[%s]", session->uid_.c_str(), op->opname_.c_str());
    base_len += snprintf(msg_buf+base_len, sizeof(msg_buf)-base_len,
                         ",\"op\":%d"
                         ",\"tsxId\":\"%s\""
                         ",\"cver\":%llu"
                         ",\"sessId\":"
                         , op->opnum_, generate_tsx_id(), session->confr_->cver_);
    bool sentSelf = false;
    for(auto o : session->confr_->sessions_){
        XSession * sess = o.second;
        if(!sendtoSelf && sess == session ){
            continue;
        }
        if(!sess->isEntered){
            continue;
        }
        if(sess->connctx_){
            int msg_len = base_len + snprintf(msg_buf+base_len, sizeof(msg_buf)-base_len, "\"%s\"}", sess->uid_.c_str());
            const char * msg = msg_buf;
            confr_mgr_send_msg(mgr, sess->connctx_, msg, msg_len);
        }
        if(sess == session){
            sentSelf = true;
        }
    }
    if(sendtoSelf && !sentSelf){
        XSession * sess = session;
        if(sess->connctx_){
            int msg_len = base_len + snprintf(msg_buf+base_len, sizeof(msg_buf)-base_len, "\"%s\"}", sess->uid_.c_str());
            const char * msg = msg_buf;
            confr_mgr_send_msg(mgr, sess->connctx_, msg, msg_len);
        }
    }
    return 0;
}

static inline
int confr_mgr_broadcast_onenter(confr_manager_t mgr, XSession * session){
    const OPPair * op = ONENTER_OP;
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"mem\":{"
                                "\"id\":\"%s\""
                                ",\"name\":\"%s\""
                            "}"
                            , session->memberId_.c_str(), session->memberName_.c_str());
        
    }
    return confr_mgr_broadcast_to_sessions(mgr, session, op, msg_buf, base_len);
}

static inline
int confr_mgr_broadcast_onexit(confr_manager_t mgr, XSession * session, int reason, bool sendtoSelf){
    const OPPair * op = ONEXIT_OP;
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"memId\":\"%s\""
                            ",\"reason\":%d"
                            , session->memberId_.c_str(), reason);
        
    }
    return confr_mgr_broadcast_to_sessions(mgr, session, op, msg_buf, base_len, sendtoSelf);
}

static inline
int confr_mgr_broadcast_onpub(confr_manager_t mgr, XStream * stream){
    const OPPair * op = ONPUB_OP;
    XSession * session = stream->rtcc_->session_;
    
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        // {id, name, voff, aoff, type, memId}
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"pubS\":{"
                                "\"id\":\"%s\""
                                ",\"name\":\"%s\""
                                ",\"voff\":%d"
                                ",\"aoff\":%d"
                                ",\"type\":%d"
                                ",\"memId\":\"%s\""
                            "}"
                            ",\"memId\":\"%s\""
                            , stream->uid_.c_str()
                            , stream->name_.c_str()
                            , stream->voff_
                            , stream->aoff_
                            , stream->type_
                            , session->memberId_.c_str()
                            , session->memberId_.c_str());
        
    }
    
    return confr_mgr_broadcast_to_sessions(mgr, stream->rtcc_->session_, op, msg_buf, base_len);
}

static inline
int confr_mgr_broadcast_onunpub(confr_manager_t mgr, XStream * stream){
    const OPPair * op = ONUNPUB_OP;
    XSession * session = stream->rtcc_->session_;
    
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        // {"op":303,"memId":"member000","pubSId":"stream-5-kdk","sessId":"sess-3-vvy"}
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"memId\":\"%s\""
                            ",\"pubSId\":\"%s\""
                            , session->memberId_.c_str(), stream->uid_.c_str());
        
    }
    
    return confr_mgr_broadcast_to_sessions(mgr, stream->rtcc_->session_, op, msg_buf, base_len);
}

static inline
int confr_mgr_broadcast_stream_ctrl(confr_manager_t mgr, XStream * stream){
    const OPPair * op = STREAMCTRL_OP;
    XSession * session = stream->rtcc_->session_;
    
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"memId\":\"%s\""
                            ",\"streamId\":\"%s\""
                            ",\"voff\":%d"
                            ",\"aoff\":%d"
                            , session->memberId_.c_str(), stream->uid_.c_str(), stream->voff_, stream->aoff_);
        
    }
    
    return confr_mgr_broadcast_to_sessions(mgr, stream->rtcc_->session_, op, msg_buf, base_len);
}

static inline
int confr_mgr_send_interest(confr_manager_t mgr, XStream * stream, int x, int y, int focus, int expo){
    const OPPair * op = INTEREST_OP;
    XSession * session = stream->rtcc_->session_;
    
    char msg_buf[256];
    int base_len = 0;
    if(base_len == 0){
        base_len = snprintf(msg_buf, sizeof(msg_buf), "{"
                            "\"memId\":\"%s\""
                            ",\"rtcId\":\"%s\""
                            ",\"x\":%d"
                            ",\"y\":%d"
                            ",\"focus\":%d"
                            ",\"expo\":%d"
                            , session->memberId_.c_str(), stream->rtcc_->remoteRtcId_.c_str(), x, y, focus, expo);
        
    }
    
    base_len += snprintf(msg_buf+base_len, sizeof(msg_buf)-base_len,
                         ",\"op\":%d"
                         ",\"tsxId\":\"%s\""
                         ",\"cver\":%llu"
                         ",\"sessId\":"
                         , op->opnum_, generate_tsx_id(), session->confr_->cver_);
    
    XSession * sess = session;
    if(sess->connctx_){
        int msg_len = base_len + snprintf(msg_buf+base_len, sizeof(msg_buf)-base_len, "\"%s\"}", sess->uid_.c_str());
        const char * msg = msg_buf;
        confr_mgr_send_msg(mgr, sess->connctx_, msg, msg_len);
    }
    return 0;
}

static inline
int confr_mgr_send_remotectrl(confr_manager_t mgr, XSession * from, XSession * to, const std::string& tsxId, const char * arg, int arg_len){
    const OPPair * op = REMOTECTRL_OP;
    StringBuffer sb;
    Writer<StringBuffer> writer(sb);
    writer.StartObject();
    writer.Key("op");
    writer.Int(op->opnum_);
    writer.Key("tsxId");
    writer.String(tsxId.c_str(), (SizeType)tsxId.length());
    writer.Key("cver");
    writer.Int(to->confr_->cver_);
    writer.Key("sessId");
    writer.String(to->uid_.c_str());
    writer.Key("memId");
    writer.String(from->memberId_.c_str());
    writer.Key("arg");
    writer.String(arg, (SizeType)arg_len);
    writer.EndObject();
    const char * msg = sb.GetString();
    int msg_len = (int)sb.GetSize();
    if(to->connctx_){
        confr_mgr_send_msg(mgr, to->connctx_, msg, msg_len);
    }
    return 0;
}



# pragma mark - entities free operations

static inline
int confr_mgr_free_rtcc(confr_manager_t mgr, XRtcc * rtcc){
    if(rtcc->pubStream_){
        XConference * confr = rtcc->session_->confr_;
        XStream * s = confr_mgr_get_stream(mgr, confr, rtcc->pubStream_->uid_, true);
        g_assert(s != NULL);
        confr_mgr_broadcast_onunpub(mgr, s);
        delete s;
    }
    
    if(rtcc->rtc_){
        rtc_endpoint_delete(rtcc->rtc_);
        rtcc->rtc_ = NULL;
    }
    
    
    mgr_dbgi("freed rtcc[%s]", rtcc->uid_.c_str());
    delete rtcc;
    
    return 0;
}

static inline
int confr_mgr_free_session(confr_manager_t mgr, XSession * session, int reason, bool sendtoSelf){
    
//    mgr_dbgi("freeing session[%s] ...", session->uid_.c_str());
    std::vector<std::string> rtcc4rm;
    for(auto o : session->rtccs_){
        XRtcc * rtcc = o.second;
        rtcc4rm.push_back(rtcc->uid_);
    }
    for(auto uid : rtcc4rm){
        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, uid, true);
        if(rtcc){
            confr_mgr_free_rtcc(mgr, rtcc);
        }
    }
    
    confr_mgr_broadcast_onexit(mgr, session, reason, sendtoSelf);
    
    mgr_dbgi("freed session[%s]", session->uid_.c_str());
    delete session;
    
    
    return 0;
}

static inline
int confr_mgr_free_confr(confr_manager_t mgr, XConference * confr){
    
//    mgr_dbgi("freeing confr[%s] ...", confr->uid_.c_str());
    std::vector<std::string> sessions4rm;
    for(auto o : confr->sessions_){
        XSession * session = o.second;
        sessions4rm.push_back(session->uid_);
    }
    for(auto uid : sessions4rm){
        XSession * session = confr_mgr_get_session(mgr, uid, true);
        if(session){
            confr_mgr_free_session(mgr, session, XSIG_REASON_DISMISS, true);
        }
    }
    
    mgr_dbgi("freed confr[%s]", confr->uid_.c_str());
    delete confr;
    
    return 0;
}

// <== entities removal operations

# pragma mark - request handle operations

static inline
void write_sdp_object( const char * sdp_type,  const char * local_sdp, int local_sdp_len, Writer<StringBuffer> &sdp_writer){
    sdp_writer.StartObject();
    sdp_writer.Key("sdp");
    sdp_writer.String(local_sdp, local_sdp_len);
    sdp_writer.Key("type");
    sdp_writer.String(sdp_type);
    sdp_writer.Key("seq");
    sdp_writer.Int(1);
    sdp_writer.EndObject();
}

static inline
void writer_candidate_object(const char * cand_sdp, const char * mid, int mlineindex, Writer<StringBuffer> &writer){
    writer.StartObject();
    writer.Key("candidate");
    writer.String(cand_sdp+2);
    
    writer.Key("mid");
    writer.String(mid);
    writer.Key("sdpMid");
    writer.String(mid);
    
    writer.Key("mlineindex");
    writer.Int(mlineindex);
    writer.Key("sdpMLineIndex");
    writer.Int(mlineindex);
    
    
    writer.Key("type");
    writer.String("candidate");
    writer.Key("seq");
    writer.Int(1);
    writer.EndObject();
}

static inline
void confr_mgr_extend_live(confr_manager_t mgr, XSession * session){
    session->isCheckLive_ = false;
}

static
int rtcmgr_set_candidate(confr_manager_t mgr, rtc_endpoint_t rtc, const Value &cand_doc, int k, std::ostringstream& oss_err){
    int ret = -1;
    do{
        Value::ConstMemberIterator it_mid = cand_doc.FindMember("mid");
        Value::ConstMemberIterator it_cand = cand_doc.FindMember("candidate");
        Value::ConstMemberIterator it_midx = cand_doc.FindMember("mlineindex");
        
        if(it_mid == cand_doc.MemberEnd()){
            it_mid = cand_doc.FindMember("sdpMid");
        }
        
        if(it_cand == cand_doc.MemberEnd()){
            it_cand = cand_doc.FindMember("sdp");
        }
        
        if(it_midx == cand_doc.MemberEnd()){
            it_midx = cand_doc.FindMember("sdpMLineIndex");
        }
        
        
        if (it_mid == cand_doc.MemberEnd() || it_cand == cand_doc.MemberEnd() || it_midx == cand_doc.MemberEnd()){
            oss_err << "candidate are non-exist some field at No." << k;
            ret = -1;
            break;
        }
        
        if(!it_mid->value.IsString() || !it_cand->value.IsString() || !it_midx->value.IsInt()){
            oss_err << "parse candidate type fail at No." << k;
            ret = -1;
            break;
        }
        
        
        std::string cand_sdp = "a=" + std::string(it_cand->value.GetString(), it_cand->value.GetStringLength());
        
        ret = rtc_endpoint_add_remote_candidate(rtc, it_midx->value.GetInt(), it_mid->value.GetString(), cand_sdp.c_str(), 0);
        if(ret){
            oss_err << "fail to add candidate sdp: " << cand_sdp;
            ret = -1;
            break;
        }
        ret = 0;
    }while(0);
    return ret;
}

static
int confr_mgr_maybe_set_candidates(confr_manager_t mgr, rtc_endpoint_t rtc, const Value &req, XProtoField& cands, std::ostringstream& oss_err){
    int ret = 0;
    if(cands.iter_ != req.MemberEnd() && cands.iter_->value.IsArray()){
        // {"candidate":"candidate:1 1 UDP 2013266431 172.17.1.58 49644 typ host","mid":"audio","mlineindex":0,"seq":1,"type":"candidate"}
        
        ret = 0;
        for(int k=0 ; k< cands.iter_->value.Size(); ++k){
            
            if(cands.iter_->value[k].IsString()){
                Document cand_doc;
                cand_doc.Parse(cands.iter_->value[k].GetString());
                if(!cand_doc.IsObject()){
                    oss_err << "parse candidate json fail at No." << k;
                    ret = -1;
                    break;
                }
                
                ret = rtcmgr_set_candidate(mgr, rtc, cand_doc, k, oss_err);
                if(ret) break;
            }else if(cands.iter_->value[k].IsObject()){
                ret = rtcmgr_set_candidate(mgr, rtc, cands.iter_->value[k], k, oss_err);
                if(ret) break;
            }else{
                oss_err << "parse candidate unknown type at No." << k;
                ret = -1;
                break;
            }
            
        }
        //        if(ret) break;
    }
    return ret;
}



static
int confr_mgr_handle_new_rtc(confr_manager_t mgr, XSession * session, Value &req
                             , XProtoInit& proto_enter, XProtoPubS& proto_pubs, XStream * sub_stream
                             , Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    rtc_endpoint_t rtc = NULL;
    do{
        std::string rtcId = MK_ITER_CSTR(proto_enter.rtcId_.iter_);
        app_config * config = app_config_get();
        bool drop_audio = false;
        bool drop_video = false;
        rtc = rtc_endpoint_create(NULL, "", 
                                  config->media_ip, config->min_port, config->max_port,
                                  config->is_dumping_rtp, config->is_dumping_rtcp,
                                  config->max_video_bitrate_kbps,
                                  config->min_video_bitrate_kbps,
                                  config->is_translating_fec,
                                  config->disable_h264,
                                  config->enable_fec,
                                  config->compensate_rtcp,
                                  config->experiment_esmb,
                                  config->lost_0,
                                  config->enable_bwe,
                                  false,
                                  true,
                                  32,
                                  drop_audio,
                                  drop_video
        );
        
        const char * sdp_type;
        if(proto_enter.sdp_.iter_ != req.MemberEnd()){
            XProtoSDP protosdp ;
            ret = protosdp.parse(proto_enter.sdp_, oss_err);
            if(ret){
                break;
            }
            
            std::string sdpType = MK_ITER_CSTR(protosdp.type_.iter_);
            if(sdpType != "offer"){
                oss_err << "expect offer sdp but got [" << sdpType << "]";
                ret = -1;
                break;
            }
            
            mgr_dbgi("server answer mode, rtcId=[%s]", rtcId.c_str());
            const Value& value = protosdp.desc_.iter_->value;
            ret = rtc_endpoint_process_offer(rtc, value.GetString(), (int)value.GetStringLength());
            if(ret){
                oss_err << "incorrect offer sdp";
                break;
            }
            ret = rtc_endpoint_generate_answer(rtc, 0);
            if(ret){
                oss_err << "error when generate answer";
                break;
            }
            sdp_type = "answer";
            
        }else {
            // TODO: server offer mode
            oss_err << "NOT support server offer mode";
            ret = -1;
            break;
            //sdp_type = "offer";
        }
        
        ret = confr_mgr_maybe_set_candidates(mgr, rtc, req, proto_enter.cands_, oss_err);
        if(ret) break;
        
        
        int local_sdp_len = 0;
        const char * local_sdp = rtc_endpoint_get_local_sdp(rtc, &local_sdp_len);
        std::vector<char *> cand_sdps;
        std::vector<const char *> cand_mids;
        std::vector<int> cand_mlineindexs;
        
        int num_mlines = rtc_endpoint_get_mline_count(rtc);
        for(int mlineindex = 0; mlineindex < num_mlines; mlineindex++){
            
            for(int stage = 0; stage <= 1; stage++){
                const char * mid = NULL;
                int candidate_count = 0;
                char **candidates = NULL;
                ret = rtc_endpoint_get_candidates(rtc, stage, mlineindex, &mid, &candidate_count, &candidates);
                if(ret){
                    mgr_dbge("rtc_endpoint_get_canidates fail, %d, stage=%d, mlineindex=%d", ret, stage, mlineindex);
                    oss_err << "get candidates fail";
                    break;
                }
                
                for(int ci = 0; ci < candidate_count; ci++){
                    cand_sdps.push_back(candidates[ci]);
                    cand_mids.push_back(mid);
                    cand_mlineindexs.push_back(mlineindex);
                }
            }
            
        }
        
        // never fail from here
        ret = 0;
        writer.Key("rtcId");
        writer.String(rtcId.c_str());
        writer.Key("sdp");
        if(session->isJsonObjMode_){
            write_sdp_object( sdp_type, local_sdp, local_sdp_len, writer);
        }else{
            StringBuffer sdp_sb;
            Writer<StringBuffer> sdp_writer(sdp_sb);
            write_sdp_object( sdp_type, local_sdp, local_sdp_len, sdp_writer);
            const char * desc = sdp_sb.GetString();
            sdp_sb.GetSize();
            writer.String(desc);
        }
        
        writer.Key("cands");
        writer.StartArray();
        for(int i = 0; i < cand_sdps.size(); i++){
            if(session->isJsonObjMode_){
                writer_candidate_object( cand_sdps[i], cand_mids[i], cand_mlineindexs[i], writer);
            }else{
                StringBuffer sdp_sb;
                Writer<StringBuffer> obj_writer(sdp_sb);
                writer_candidate_object( cand_sdps[i], cand_mids[i], cand_mlineindexs[i], obj_writer);
                const char * desc = sdp_sb.GetString();
                sdp_sb.GetSize();
                writer.String(desc);
            }
        }
        writer.EndArray();
        
        XRtcc * rtcc = confr_mgr_new_rtcc(mgr, session, rtcId);
        rtcc->rtc_ = rtc;
        rtcc->remoteRtcId_ = rtcId;
        rtc = NULL;
        
        mgr_dbgi("bind rtcc [%s] -> local[%s] -> remote[%s]"
                 , rtcc->uid_.c_str()
                 , rtc_endpoint_get_id(rtcc->rtc_)
                 , rtcc->remoteRtcId_.c_str());
        
        if(proto_enter.pubS_.iter_ != req.MemberEnd()){
            XStream * pubstream = confr_mgr_new_stream(mgr, rtcc);
            const Value& pubvalue = proto_enter.pubS_.iter_->value;
            if(proto_pubs.type_.iter_ != pubvalue.MemberEnd()){
                pubstream->type_ = proto_pubs.type_.iter_->value.GetInt();
            }
            if(proto_pubs.name_.iter_ != pubvalue.MemberEnd()){
                pubstream->name_ = MK_ITER_CSTR(proto_pubs.name_.iter_);
            }
            if(proto_pubs.voff_.iter_ != pubvalue.MemberEnd()){
                pubstream->voff_ = proto_pubs.voff_.iter_->value.GetInt();
            }
            if(proto_pubs.aoff_.iter_ != pubvalue.MemberEnd()){
                pubstream->aoff_ = proto_pubs.aoff_.iter_->value.GetInt();
            }
            
            mgr_dbgi("pub stream [%s]->[%s]", rtcc->remoteRtcId_.c_str(), rtcc->pubStream_->uid_.c_str());
            
            confr_mgr_broadcast_onpub(mgr, rtcc->pubStream_);
            writer.Key("streamId");
            writer.String(rtcc->pubStream_->uid_.c_str());
            
        }
        
        if(proto_enter.subSId_.iter_ != req.MemberEnd()){
            XStream * ss = confr_mgr_sub_stream(mgr, sub_stream, rtcc);
            if(!ss){
                ret = -1;
                oss_err << "rtcc already subsr, [" << rtcc->uid_ << "]->[" << sub_stream->uid_ << "]";
                break;
            }

            rtc_endpoint_subscribe(rtcc->rtc_, rtcc->subStream_->rtcc_->rtc_, true, true);
            mgr_dbgi("sub stream [%s]->[%s]", rtcc->remoteRtcId_.c_str(), rtcc->subStream_->uid_.c_str());
        }

        ret = 0;
    }while(0);
    
    if(rtc){
        rtc_endpoint_delete(rtc);
        rtc = NULL;
    }
    
    return ret;
}

static inline
int confr_mgr_write_cver(confr_manager_t mgr, XSession * session, Writer<StringBuffer> &writer){
    writer.Key("cver");
    writer.Int64(session->confr_->cver_);
    return 0;
}

static inline
int confr_mgr_write_confr(confr_manager_t mgr, XSession * session, Writer<StringBuffer> &writer){
    writer.Key("mems");
    writer.StartArray();
    for(auto iter : session->confr_->sessions_){
        // {id :m001, memName: yss}
        XSession * sess = iter.second;
        writer.StartObject();
        writer.Key("id");
        writer.String(sess->memberId_.c_str());
        writer.Key("name");
        writer.String(sess->memberName_.c_str());
        writer.EndObject();
    }
    writer.EndArray();
    
    writer.Key("streams");
    writer.StartArray();
    for(auto iter : session->confr_->streams_){
        // {id, name, voff, aoff, type, memId}
        XStream * stream = iter.second;
        writer.StartObject();
        writer.Key("id");
        writer.String(stream->uid_.c_str());
        writer.Key("name");
        writer.String(stream->name_.c_str());
        writer.Key("voff");
        writer.Int(stream->voff_);
        writer.Key("aoff");
        writer.Int(stream->aoff_);
        writer.Key("type");
        writer.Int(stream->type_);
        writer.Key("memId");
        writer.String(stream->rtcc_->session_->memberId_.c_str());
        writer.EndObject();
    }
    writer.EndArray();
    
    confr_mgr_write_cver(mgr, session, writer);
    
    return 0;
}



static
int confr_mgr_handle_init_common(confr_manager_t mgr, void * conn_ctx , const char * conn_id, Value &req, XProtoInit& proto_req
                           , Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    XSession * alloc_session = NULL;
    XConference * alloc_confr = NULL;
    do{
//        XProtoEnter proto_req;
//        ret = proto_req.parse(req, oss_err);
//        if(ret){
//            break;
//        }
        
        XProtoTicket prototkt;
        if(proto_req.tkt_.iter_ != req.MemberEnd()){
            ret = prototkt.parse(proto_req.tkt_, oss_err);
            if(ret){
                break;
            }
        }
        
        XProtoPubS proto_pubs;
        if(proto_req.pubS_.iter_ != req.MemberEnd()){
            ret = proto_pubs.parse(proto_req.pubS_, oss_err);
            if(ret){
                break;
            }
        }
        
        XSession * session = NULL;
        XConference * confr = NULL;
        
        if(proto_req.sessId_.iter_ != req.MemberEnd()){
            std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
            session = confr_mgr_get_session(mgr, sessionId, false);
            if(!session){
                oss_err << "non-exist session [" << sessionId << "]";
                ret = -1;
                break;
            }
            confr_mgr_bind_session_conn(mgr, session, conn_id, conn_ctx);
            confr = session->confr_;
        }else{
            std::string confrId = MK_ITER_CSTR(prototkt.confrId_.iter_);
            confr = confr_mgr_get_confr(mgr, confrId, false);
            if(!confr){
                alloc_confr = confr_mgr_new_confr(mgr, confrId);
                confr = alloc_confr;
            }
        }
        
        XStream * sub_stream = NULL;
        if(proto_req.subSId_.iter_ != req.MemberEnd()){
            std::string subSId = MK_ITER_CSTR(proto_req.subSId_.iter_);
            sub_stream = confr_mgr_get_stream( mgr, confr, subSId, false);
            if(!sub_stream){
                oss_err << "subsr to non-exit stream [" << subSId << "]";
                ret = -1;
                break;
            }
        }
        
        if(!session){
            alloc_session = confr_mgr_new_session(mgr, confr);
            session = alloc_session;
            session->memberId_ = generate_member_id();
            session->memberName_ = MK_ITER_CSTR(prototkt.memName_.iter_);
            if(proto_req.tkt_.iter_->value.IsObject()){
                session->isJsonObjMode_ = true;
            }
            
            confr_mgr_bind_session_conn(mgr, session, conn_id, conn_ctx);
            mgr_dbgi("bind memberId [%s]-[%s]->[%s]", session->uid_.c_str(), session->memberId_.c_str(), session->memberName_.c_str());
            
            XSession * oldSession = confr_mgr_add_member(mgr, session);
            if(oldSession){
                mgr_dbgi("kick old session [%s]-[%s]", oldSession->memberName_.c_str(), oldSession->uid_.c_str());
                confr_mgr_free_session(mgr, oldSession, XSIG_REASON_OTHER_DEVICE, true);
            }
            
            confr_mgr_broadcast_onenter(mgr, session);
        }

        
        if(proto_req.rtcId_.iter_ != req.MemberEnd()){
            ret = confr_mgr_handle_new_rtc(mgr, session, req, proto_req, proto_pubs, sub_stream, writer, oss_err);
            if(ret) break;
        }
        
        
        confr_mgr_write_confr(mgr, session, writer);
        
        writer.Key("sessId");
        writer.String(session->uid_.c_str());
        
        if(alloc_session){
            writer.Key("memId");
            writer.String(session->memberId_.c_str());
        }

        if(alloc_session){
            alloc_session->isEntered = true;
        }
        
        alloc_session = NULL;
        alloc_confr = NULL;
        confr_mgr_extend_live(mgr, session);
        
        ret = 0;
    }while(0);
    
    if(alloc_session){
        XSession * sess = confr_mgr_get_session(mgr, alloc_session->uid_, true);
        g_assert(sess == alloc_session);
        ret = confr_mgr_free_session(mgr, sess, XSIG_REASON_NORMAL, false);
        alloc_session = NULL;
    }
    
    if(alloc_confr){
        confr_mgr_get_confr(mgr, alloc_confr->uid_, true);
        ret = confr_mgr_free_confr(mgr, alloc_confr);
        alloc_confr = NULL;
    }
    
    return ret;
}

static
int confr_mgr_handle_enter(confr_manager_t mgr, void * conn_ctx , const char * conn_id, Value &req
                                 , Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoEnter proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        
        ret = confr_mgr_handle_init_common(mgr, conn_ctx, conn_id, req, proto_req, writer, oss_err);

    }while(0);
    return ret;
}

static
int confr_mgr_handle_initC(confr_manager_t mgr, void * conn_ctx , const char * conn_id, Value &req
                           , Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoInitC proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        
        ret = confr_mgr_handle_init_common(mgr, conn_ctx, conn_id, req, proto_req, writer, oss_err);
        
    }while(0);
    return ret;
}



static
int confr_mgr_handle_tcklC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoTcklC proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        std::string rtcId = MK_ITER_CSTR(proto_req.rtcId_.iter_);
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        
        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, confr_mgr_make_rtcc_uid(rtcId), false);
        if(!rtcc){
            oss_err << "non-exist rtc [" << rtcId << "]";
            ret = -1;
            break;
        }
        
        ret = confr_mgr_maybe_set_candidates(mgr, rtcc->rtc_, req, proto_req.cands_, oss_err);
        if(ret) break;
        
        writer.Key("sessId");
        writer.String(session->uid_.c_str());
        confr_mgr_extend_live(mgr, session);
        
        ret = 0;
    }while(0);
    
    return ret;
}


static
int confr_mgr_handle_termC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoTermC proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        std::string rtcId = MK_ITER_CSTR(proto_req.rtcId_.iter_);
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        
        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, confr_mgr_make_rtcc_uid(rtcId), true);
        if(!rtcc){
            oss_err << "non-exist rtc [" << rtcId << "]";
            ret = -1;
            break;
        }
        
        ret = confr_mgr_free_rtcc(mgr, rtcc);
        if(ret) {
            oss_err << "error free rtc[" << rtcId << "]";
            ret = -1;
            break;
        }
        
        writer.Key("sessId");
        writer.String(session->uid_.c_str());
        
        confr_mgr_extend_live(mgr, session);
        
        ret = 0;
    }while(0);
    
    
    return ret;
}

static
int confr_mgr_handle_exit(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoExit proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, true);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        int reason = XSIG_REASON_NORMAL;
        if(proto_req.reason_.iter_ != req.MemberEnd()){
            reason = MK_ITER_INT(proto_req.reason_.iter_);
        }
        
        
        XConference * confr = session->confr_;
        ret = confr_mgr_free_session(mgr, session, reason, false);
        if(ret) {
            oss_err << "error free session[" << sessionId << "]";
            ret = -1;
            break;
        }
        
        if(confr->sessions_.size() == 0){
            std::string confrId = confr->uid_;
            confr_mgr_get_confr(mgr, confrId, true);
            ret = confr_mgr_free_confr(mgr, confr);
            if(ret) {
                oss_err << "error free confr[" << confrId << "]";
                ret = -1;
                break;
            }
        }
        
        writer.Key("sessId");
        writer.String(sessionId.c_str());
        
        ret = 0;
    }while(0);
    
    
    return ret;
}

// TODO: subC and unsubC

//static
//int confr_mgr_handle_subC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
//    int ret = -1;
//    do{
//        XProtoSubC proto_req;
//        ret = proto_req.parse(req, oss_err);
//        if(ret){
//            break;
//        }
//        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
//        std::string rtcId = MK_ITER_CSTR(proto_req.rtcId_.iter_);
//        std::string subSId = MK_ITER_CSTR(proto_req.subSId_.iter_);
//        
//        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
//        if(!session){
//            oss_err << "non-exist session [" << sessionId << "]";
//            ret = -1;
//            break;
//        }
//        
//        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, confr_mgr_make_rtcc_uid(rtcId), true);
//        if(!rtcc){
//            oss_err << "non-exist rtc [" << rtcId << "]";
//            ret = -1;
//            break;
//        }
//        
//        XStream * stream = confr_mgr_get_stream(mgr, session->confr_, subSId, false);
//        if(!stream){
//            oss_err << "subsr to non-exit stream [" << subSId << "]";
//            ret = -1;
//            break;
//        }
//        
//        if(rtcc->subStream_){
//            if(rtcc->subStream_ != stream){
//                oss_err << "dup subsr, exist [" << rtcId << "]->[" << rtcc->subStream_->uid_ << "]";
//                ret = -1;
//                break;
//            }
//            else{
//                mgr_dbgi("already subsr [%s]->[%s]", rtcId.c_str(), subSId.c_str());
//                ret = 0;
//                break;
//            }
//        }
//        
//        confr_mgr_sub_stream(mgr, stream, rtcc);
//        rtc_endpoint_subscribe(rtcc->rtc_, rtcc->subStream_->rtcc_->rtc_);
//        mgr_dbgi("sub stream [%s]->[%s]", rtcc->remoteRtcId_.c_str(), rtcc->subStream_->uid_.c_str());
//        
//        writer.Key("sessId");
//        writer.String(session->uid_.c_str());
//        
//        ret = 0;
//    }while(0);
//    
//    
//    return ret;
//}
//
//
//static
//int confr_mgr_handle_unsubC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
//    int ret = -1;
//    do{
//        XProtoUnsubC proto_req;
//        ret = proto_req.parse(req, oss_err);
//        if(ret){
//            break;
//        }
//        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
//        std::string rtcId = MK_ITER_CSTR(proto_req.rtcId_.iter_);
//        std::string subSId = MK_ITER_CSTR(proto_req.subSId_.iter_);
//        
//        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
//        if(!session){
//            oss_err << "non-exist session [" << sessionId << "]";
//            ret = -1;
//            break;
//        }
//        
//        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, confr_mgr_make_rtcc_uid(rtcId), true);
//        if(!rtcc){
//            oss_err << "non-exist rtc [" << rtcId << "]";
//            ret = -1;
//            break;
//        }
//        
//        if(rtcc->subStreamId_ != subSId){
//            oss_err << "inconsistent unsub [" << rtcId << "]: "
//                << "[" << rtcc->subStreamId_ << "] != [" << subSId << "]";
//            ret = -1;
//            break;
//        }
//        
//        
//        if(!rtcc->subStream_){
//            mgr_dbgi("unsub stream emtpy [%s]->[%s]", rtcc->remoteRtcId_.c_str(), rtcc->subStream_->uid_.c_str());
//            rtcc->subStreamId_ = "";
//        }else{
//            mgr_dbgi("unsub stream [%s]->[%s]", rtcc->remoteRtcId_.c_str(), rtcc->subStream_->uid_.c_str());
//            confr_mgr_unsub_stream(mgr, rtcc->subStream_, rtcc);
//            rtc_endpoint_unsubscribe(rtcc->rtc_, rtcc->subStream_->rtcc_->rtc_);
//        }
//        
//        writer.Key("sessId");
//        writer.String(session->uid_.c_str());
//        
//        ret = 0;
//    }while(0);
//    
//    
//    return ret;
//}

static
int confr_mgr_handle_getC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoGetC proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        
        int64_t cver = proto_req.cver_.iter_->value.GetInt64();
        if(cver == session->confr_->cver_){
            confr_mgr_write_cver(mgr, session, writer);
        }else{
            confr_mgr_write_confr(mgr, session, writer);
        }
        
        writer.Key("sessId");
        writer.String(session->uid_.c_str());
        
        confr_mgr_extend_live(mgr, session);
        
        ret = 0;
    }while(0);
    
    
    return ret;
}

static
int confr_mgr_handle_stream_ctrl(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoStreamCtrl proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        std::string rtcId = MK_ITER_CSTR(proto_req.rtcId_.iter_);
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, confr_mgr_make_rtcc_uid(rtcId), false);
        if(!rtcc){
            oss_err << "non-exist rtc [" << rtcId << "]";
            ret = -1;
            break;
        }
        if(!rtcc->pubStream_){
            oss_err << "NOT-pub rtc [" << rtcId << "]";
            ret = -1;
            break;
        }
        
        if(proto_req.voff_.iter_ != req.MemberEnd()){
            rtcc->pubStream_->voff_ = MK_ITER_INT(proto_req.voff_.iter_);
        }else{
            rtcc->pubStream_->voff_ = 0;
        }
        
        if(proto_req.aoff_.iter_ != req.MemberEnd()){
            rtcc->pubStream_->aoff_ = MK_ITER_INT(proto_req.aoff_.iter_);
        }else{
            rtcc->pubStream_->aoff_ = 0;
        }
        
        ret = confr_mgr_broadcast_stream_ctrl(mgr, rtcc->pubStream_);
        
        ret = 0;
    }while(0);
    
    
    return ret;
}

static
int confr_mgr_handle_interestC(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoInterest proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        std::string streamId = MK_ITER_CSTR(proto_req.streamId_.iter_);
        int x = MK_ITER_INT(proto_req.x_.iter_);
        int y = MK_ITER_INT(proto_req.y_.iter_);
        int focus = 1;
        int expo = 1;
        if(proto_req.focus_.iter_ != req.MemberEnd()){
            focus = MK_ITER_INT(proto_req.focus_.iter_);
        }
        if(proto_req.expo_.iter_ != req.MemberEnd()){
            expo = MK_ITER_INT(proto_req.expo_.iter_);
        }
        
        XSession * session = confr_mgr_get_session(mgr, sessionId, false);
        if(!session){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        
        XStream * stream = confr_mgr_get_stream( mgr, session->confr_, streamId, false);
        if(!stream){
            oss_err << "interest in non-exit stream [" << streamId << "]";
            ret = -1;
            break;
        }
    
        
        ret = confr_mgr_send_interest(mgr, stream, x, y, focus, expo);
        
        ret = 0;
    }while(0);
    
    
    return ret;
}

static
int confr_mgr_handle_remotectrl(confr_manager_t mgr, Value &req, Writer<StringBuffer> &writer, std::ostringstream& oss_err){
    int ret = -1;
    do{
        XProtoRemoteCtrl proto_req;
        ret = proto_req.parse(req, oss_err);
        if(ret){
            break;
        }
        std::string sessionId = MK_ITER_CSTR(proto_req.sessId_.iter_);
        std::string memberId = MK_ITER_CSTR(proto_req.memId_.iter_);
        std::string tsxId = MK_ITER_CSTR(proto_req.tsxId_.iter_);
        XSession * from = confr_mgr_get_session(mgr, sessionId, false);
        if(!from){
            oss_err << "non-exist session [" << sessionId << "]";
            ret = -1;
            break;
        }
        
        XSession * to = confr_mgr_get_session_by_member_id(mgr, from->confr_, memberId, false);
        if(!to){
            oss_err << "non-exist to member [" << memberId << "]";
            ret = -1;
            break;
        }
        
        ret = confr_mgr_send_remotectrl(mgr, from, to, tsxId
                                        , proto_req.arg_.iter_->value.GetString(), proto_req.arg_.iter_->value.GetStringLength());
        
        ret = 0;
    }while(0);
    
    return ret;
}

// <== request handle operations



# pragma mark - confr_mgr public

int confr_mgr_recv_msg(confr_manager_t mgr
                       , void * conn_ctx , const char * conn_id
                       , const char * json, int json_len
                       , uint8_t * rsp_buf, uint32_t * rsp_length){
    int ret = -1;
    Document req_doc;
    const OPPair * op = NULL;
    const OPPair *rspop = NULL;
    int is_op_string = 0;
    std::ostringstream oss_err;
    StringBuffer sb;
    Writer<StringBuffer> writer(sb);
    writer.StartObject();
    
    do{
        *rsp_length = 0;
        
        req_doc.Parse(json);
        if(!req_doc.IsObject()){
            mgr_dbgi("signal msg: req ==> %.*s\n", json_len, json);
            oss_err << "req json is NOT object";
            ret = -1;
            break;
        }
        
        Value::ConstMemberIterator it_op = req_doc.FindMember("op");
        if(it_op == req_doc.MemberEnd() || (!it_op->value.IsString() && !it_op->value.IsInt())){
            mgr_dbgi("signal msg: req ==> %.*s\n", json_len, json);
            oss_err << "get field op error";
            ret = -1;
            break;
        }
        
        if(it_op->value.IsString()){
            is_op_string = 1;
            op = OPPair::get(std::string(it_op->value.GetString(), it_op->value.GetStringLength()));
        }else{
            op = OPPair::get(it_op->value.GetInt());
        }
        if(!op){
            mgr_dbgi("signal msg: req ==> %.*s\n", json_len, json);
            oss_err << "unknown op";
            ret = -1;
            break;
        }
        
        if(!op->ismute_){
            mgr_dbgi("signal msg: req ==> %.*s\n", json_len, json);
        }
        
        // write rsp op
        rspop = OPPair::get(XSIG_OP_RSP);
        writer.Key("op");
        if(is_op_string){
            writer.String(rspop->opname_.c_str());
        }else{
            writer.Int(rspop->opnum_);
        }
        
        
        if(op->opnum_ == XSIG_OP_ENTER){
            ret = confr_mgr_handle_enter(mgr, conn_ctx, conn_id, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_INITC){
            ret = confr_mgr_handle_initC(mgr, conn_ctx, conn_id, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_TCKLC){
            ret = confr_mgr_handle_tcklC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_TERMC){
            ret = confr_mgr_handle_termC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_EXIT){
            ret = confr_mgr_handle_exit(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_GETC){
            ret = confr_mgr_handle_getC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_PINGC){
            ret = confr_mgr_handle_getC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_SUBC){
//            ret = confr_mgr_handle_subC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_USUBC){
//            ret = confr_mgr_handle_unsubC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_STREAM_CONTROL){
            ret = confr_mgr_handle_stream_ctrl(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_INTERESTC){
            ret = confr_mgr_handle_interestC(mgr, req_doc, writer, oss_err);
        }else if(op->opnum_ == XSIG_OP_RCTRLC){
            ret = confr_mgr_handle_remotectrl(mgr, req_doc, writer, oss_err);
        }
        else{
            oss_err << "unhandle op [" <<  op->opnum_ << "]->[" << op->opname_.c_str() << "]";
            ret = -1;
            break;
        }
        if(ret) break;
        

        
        ret = 0;
    }while(0);
    
    if(*rsp_length == 0 ){
        if(!rspop){
            rspop = OPPair::get(XSIG_OP_RSP);
            writer.Key("op");
            if(is_op_string){
                writer.String(rspop->opname_.c_str());
            }else{
                writer.Int(rspop->opnum_);
            }
        }
        
        writer.Key("result");
        writer.Int(ret);
        
        if(req_doc.IsObject()){
            Value::ConstMemberIterator it_tsxId = req_doc.FindMember("tsxId");
            if(it_tsxId != req_doc.MemberEnd() && it_tsxId->value.IsString()){
                writer.Key("tsxId");
                writer.String(it_tsxId->value.GetString(), it_tsxId->value.GetStringLength());
            }
        }
        
        std::string error = oss_err.str();
        if(error.length() > 0){
            mgr_dbge("%s", error.c_str());
            writer.Key("errmsg");
            writer.String(error.c_str());
        }
        
        writer.EndObject();
        
        memcpy(rsp_buf, sb.GetString(),  sb.GetSize());
        *rsp_length = sb.GetSize();
        rsp_buf[*rsp_length] = '\0';
    }
    
    if(!op || !op->ismute_){
        mgr_dbgi("signal msg: rsp <== %.*s\n", *rsp_length, rsp_buf);
    }
    
    
    return ret;
}

int confr_mgr_update_conn(confr_manager_t mgr, void * conn_ctx , const char * conn_id){
    XSession * session = confr_mgr_get_session_conn(mgr, conn_id);
    if(session){
        confr_mgr_bind_session_conn(mgr, session, conn_id, conn_ctx);
    }
    return 0;
}

static inline
bool confr_mgr_check_session_live(confr_manager_t mgr, XSession * session, int64_t now){
    std::vector<std::string> dead_rtccs;
    bool is_live = false;
    for(auto o : session->rtccs_){
        XRtcc * rtcc = o.second;
        if(rtcc->rtc_){
            if(rtc_endpoint_check_live(rtcc->rtc_, now)){
                is_live = true;
            }else{
                dead_rtccs.push_back(rtcc->uid_);
            }
        }
    }
    for(auto uid : dead_rtccs){
        XRtcc * rtcc = confr_mgr_get_rtcc(mgr, session, uid, true);
        if(rtcc){
            confr_mgr_free_rtcc(mgr, rtcc);
        }
    }
    if(is_live){
        session->isCheckLive_ = true;
        session->lastCheckLiveTime = now;
        return true;
    }
    
    const int64_t timeout_ms = 15*1000;
    if(session->rtccs_.size() > 0 || dead_rtccs.size() == 0){
        if(!session->isCheckLive_){
            session->isCheckLive_ = true;
            session->lastCheckLiveTime = now;
            return true;
        }
        
        int64_t elapse = (now-session->lastCheckLiveTime);
        if(elapse < timeout_ms){
            return true;
        }
    }

    mgr_dbgi("timeout session [%s] in milli [%lld]", session->uid_.c_str(), timeout_ms);
    return false;
}

int confr_mgr_timer(confr_manager_t mgr, int64_t now){
    std::vector<std::string> dead_confrs;
    std::vector<std::string> dead_sessions;
    for(auto oc : *mgr->confrs) {
        XConference * confr = oc.second;
        int num_die_sessions = 0;
        bool die = true;
        for(auto os : confr->sessions_){
            XSession * sess = os.second;
            if(confr_mgr_check_session_live(mgr, sess, now)){
                die = false;
            }else{
                dead_sessions.push_back(sess->uid_);
                ++num_die_sessions;
            }
            
        }
        if(die){
            if(num_die_sessions == 0){
                if(!confr->isCheckLive_){
                    confr->isCheckLive_ = true;
                    confr->lastCheckLiveTime = now;
                    continue;
                }
                
                const int64_t timeout_ms = 15*1000;
                int64_t elapse = (now-confr->lastCheckLiveTime);
                if(elapse < timeout_ms){
                    continue;
                }
                mgr_dbgi("timeout confr [%s] in milli [%lld]", confr->uid_.c_str(), timeout_ms);
            }
            dead_confrs.push_back(confr->uid_);
        }else{
            confr->isCheckLive_ = true;
            confr->lastCheckLiveTime = now;
        }
        
    }
    for(auto uid : dead_sessions){
        XSession * sess = confr_mgr_get_session(mgr, uid, true);
        if(sess){
            confr_mgr_free_session(mgr, sess, XSIG_REASON_TIMEOUT, true);
        }
    }
    for(auto uid : dead_confrs){
        XConference * confr = confr_mgr_get_confr(mgr, uid, true);
        if(confr){
            confr_mgr_free_confr(mgr, confr);
        }
    }
    
    return 0;
}

confr_manager_t confr_mgr_create(void * main_ctx, confr_mgr_callbacks * callbacks){
    confr_manager_t mgr = NULL;
    ZERO_ALLOC_(mgr, confr_manager_t, sizeof(struct confr_manager_st));
    mgr->callbacks = *callbacks;
    mgr->callbacks.main_ctx = main_ctx;
    mgr->confrs = new XConferenceMap;
    mgr->sessions = new XSessionMap;
    mgr->conn_sessions = new XSessionMap;
    return mgr;
}

void confr_manager_delete(confr_manager_t mgr){
    if(!mgr) return;
    
    std::vector<std::string> confrIds;
    for(auto o : *mgr->confrs){
        confrIds.push_back(o.first);
    }
    for(auto uid : confrIds){
        XConference * confr = confr_mgr_get_confr(mgr, uid, true);
        if(confr){
            confr_mgr_free_confr(mgr, confr);
        }
    }
    
    if(mgr->conn_sessions){
        delete mgr->conn_sessions;
        mgr->conn_sessions = NULL;
    }
    
    if(mgr->sessions){
        delete mgr->sessions;
        mgr->sessions = NULL;
    }
    
    if(mgr->confrs){
        delete mgr->confrs;
        mgr->confrs = NULL;
    }
    
    free(mgr);
}


# pragma mark - rtc_signal_server


#define conn_dbgv(FMT, ARGS...) dbgv("|%s| " FMT, conn->obj_id, ##ARGS )
#define conn_dbgi(FMT, ARGS...) dbgi("|%s| " FMT, conn->obj_id, ##ARGS )
#define conn_dbge(FMT, ARGS...) dbge("|%s| " FMT, conn->obj_id, ##ARGS )

typedef struct rtc_signal_server_st * rtc_signal_server_t;
typedef struct rtc_signal_conn_st * rtc_signal_conn_t;


struct rtc_signal_conn_st{
    char obj_id[MAX_OBJ_ID_LEN];
    rtc_signal_server_t owner;
    std::list<rtc_signal_conn_t>::iterator ref;
    SoupWebsocketConnection * websocket;
};


struct rtc_signal_server_st{
    std::list<rtc_signal_conn_t> * conn_list; // connection list
    SoupServer *soup_server;
    confr_manager_t mgr;
    uint8_t rsp_buf[32*1024];
    GSource * timer;
};



static void rtc_signal_conn_delete(rtc_signal_conn_t conn);

static
void on_connection_text_message (SoupWebsocketConnection *ws,
                                 SoupWebsocketDataType type,
                                 GBytes *message,
                                 gpointer user_data){
    rtc_signal_conn_t conn = (rtc_signal_conn_t) user_data;
    
    if(type != SOUP_WEBSOCKET_DATA_TEXT){
        return;
    }
    
    gsize sz;
    const char *contents = (const char *)g_bytes_get_data (message, &sz);
    int len = sz;
//    conn_dbgi("on_connection_text_message (%d bytes): [%.*s]", len, len, contents);
    uint32_t rsp_len = 0;
    confr_mgr_recv_msg(conn->owner->mgr, conn, conn->obj_id, contents, len, conn->owner->rsp_buf, &rsp_len);
    if(rsp_len > 0){
        soup_websocket_connection_send_text (ws, (char*)conn->owner->rsp_buf);
    }
    
//    const char * msg = "{\"result\":-1}";
//    conn_dbgi("send messmage: %s", msg);
//    soup_websocket_connection_send_text (ws, msg);
    
}

static
void on_connection_closed (SoupWebsocketConnection *ws, gpointer user_data){
    rtc_signal_conn_t conn = (rtc_signal_conn_t) user_data;
    rtc_signal_conn_delete(conn);
}


static
rtc_signal_conn_t rtc_signal_conn_create(rtc_signal_server_t owner, SoupWebsocketConnection * connection){
    rtc_signal_conn_t conn = NULL;
    ZERO_ALLOC_(conn, rtc_signal_conn_t, sizeof(struct rtc_signal_conn_st));
//    snprintf(conn->obj_id, sizeof(conn->obj_id), "ws-%lld", get_next_conn_id());
    generate_obj_id("conn", conn->obj_id, sizeof(conn->obj_id));
    conn->owner = owner;
    conn->ref = owner->conn_list->insert(owner->conn_list->end(), conn);
    conn->websocket = (SoupWebsocketConnection *)g_object_ref (connection);
    g_signal_connect (conn->websocket, "message", G_CALLBACK (on_connection_text_message), (gpointer)conn);
    g_signal_connect (conn->websocket, "closed", G_CALLBACK (on_connection_closed), (gpointer)conn);
    
    SoupURI * uri = soup_websocket_connection_get_uri (conn->websocket);
    char * str = soup_uri_to_string (uri, FALSE);
    conn_dbgi("websocket connected, at [%s], [%s]->[%p]", str, conn->obj_id, conn);
    g_free (str);
//    soup_uri_free (uri);
    
    return conn;
}

static
void rtc_signal_conn_delete(rtc_signal_conn_t conn){
    if(!conn) return;
    conn_dbgi("websocket closed, [%s]->[%p]", conn->obj_id, conn);
    confr_mgr_update_conn(conn->owner->mgr, NULL, conn->obj_id);
    
    conn->owner->conn_list->erase(conn->ref);
    if(conn->websocket){
        g_object_unref ( conn->websocket);
        conn->websocket = NULL;
    }
    
    free(conn);
}


void rtc_signal_server_delete(rtc_signal_server_t obj){
    if(!obj) return;

    if(obj->timer != NULL) {
        g_source_destroy(obj->timer);
        g_source_unref(obj->timer);
        obj->timer = NULL;
    }
    
    if(obj->conn_list){
        while(!obj->conn_list->empty()){
            rtc_signal_conn_t conn = obj->conn_list->front();
            rtc_signal_conn_delete(conn);
        }
        delete obj->conn_list;
        obj->conn_list = NULL;
    }
    
    if(obj->mgr){
        confr_manager_delete(obj->mgr);
        obj->mgr = NULL;
    }
    
    free(obj);
}




static
void got_server_connection (SoupServer *server,
                            SoupWebsocketConnection *connection,
                            const char *path,
                            SoupClientContext *client,
                            gpointer user_data){
    rtc_signal_server_t obj = (rtc_signal_server_t) user_data;
    gdbgi("got_server_connection: path=[%s], ws=%p", path, connection);
    rtc_signal_conn_create(obj, connection);
}

static
gboolean on_timer_handler(gpointer user_data){
    rtc_signal_server_t obj = (rtc_signal_server_t) user_data;
    int64_t now = get_timestamp_ms();
    confr_mgr_timer(obj->mgr, now);
    return TRUE;
}


static
void mgr_callback_on_send(confr_manager_t mgr, void * main_ctx, void * conn_ctx, const char * msg, int msg_len){
    rtc_signal_conn_t conn = (rtc_signal_conn_t) conn_ctx;
    g_assert(conn != NULL);
    soup_websocket_connection_send_text (conn->websocket, msg);
}

static confr_mgr_callbacks mgr_callbacks = {
    .main_ctx = NULL,
    .on_send = mgr_callback_on_send
};



#define BREAK_IF_ERROR(reason, error) \
    if (error != NULL){\
        gdbge ("%s:%s", reason, error->message);\
        g_clear_error (&error);\
        ret = -1;\
        break;\
    }


rtc_signal_server_t rtc_signal_server_create(const char * listen_ip, int listen_port){

    int ret = 0;
    rtc_signal_server_t obj = NULL;
    GSocketAddress *addr;
    GError *error = NULL;
    
    do{
        ZERO_ALLOC_(obj, rtc_signal_server_t, sizeof(struct rtc_signal_server_st));
        obj->conn_list = new std::list<rtc_signal_conn_t>;
        obj->mgr = confr_mgr_create(obj, &mgr_callbacks);
        obj->soup_server = soup_server_new(NULL, NULL);
        obj->timer = g_timeout_source_new(2000);
        g_source_set_callback(obj->timer, on_timer_handler, obj, NULL);
        g_source_attach(obj->timer, NULL);

        
        addr = g_inet_socket_address_new_from_string (listen_ip, listen_port);
        BREAK_IF_ERROR ("new address", error);
        
        
        obj->soup_server = soup_server_new(NULL, NULL);
        soup_server_listen (obj->soup_server,
                            addr,
                            (SoupServerListenOptions)0,
                            &error);
        BREAK_IF_ERROR ("soup_server_listen", error);
        g_object_unref (addr);
        
        
        const char *path = NULL; // "/"
        const char * origin = NULL;
        char ** protocols = NULL;
        soup_server_add_websocket_handler (obj->soup_server, path,
                                           origin, (char **) protocols,
                                           got_server_connection, obj, NULL);
        
        
        // print listen interface
        GSList *uris = soup_server_get_uris (obj->soup_server);
        GSList *u;
        for (u = uris; u; u = u->next) {
            char * str = soup_uri_to_string ((SoupURI *)u->data, FALSE);
            gdbgi ("websocket service at %s\n", str);
            g_free (str);
            soup_uri_free ((SoupURI *)u->data);
        }
        g_slist_free (uris);
        
        
        ret = 0;
        
    }while(0);
    if(ret){
        rtc_signal_server_delete(obj);
        obj = NULL;
    }
    return obj;
}


int websocket_server_main( int argc, char *argv[] ){
    app_config * config = app_config_get();
    rtc_signal_server_t server = rtc_signal_server_create(config->http_ip, config->http_port);
    if(!server){
        return 2;
    }

    GMainLoop * mainloop = g_main_loop_new(NULL, FALSE);
    g_main_loop_run(mainloop);
    
    return 0;
}

// 3 header for inet_aton
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

static
int check_localip(const char *localip){
    //    localip = "172.17.1.197";
    struct sockaddr_in sin = { 0 };
    int sock = -1;
    int ret = -1;
    
    do{
        sin.sin_family = AF_INET;
        sin.sin_port = htons(0);
        inet_aton(localip, &sin.sin_addr);
        sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
        if(sock < 0){
            ret = -111;
            break;
        }
        
        ret = bind(sock, (struct sockaddr *) &sin, sizeof(sin));
        if(ret < 0){
            ret = -222;
            break;
        }
        
        ret = 0;
    }while(0);
    
    if(sock >= 0){
        close(sock);
    }
    
    return ret;
}

static
void dump_config(app_config * config){
    gdbgi("  opt: log-config-file=[%s]", config->log_config_file_name);
    gdbgi("  opt: media-ip=[%s]", config->media_ip);
    gdbgi("  opt: min-port=[%d]", config->min_port);
    gdbgi("  opt: max-port=[%d]", config->max_port);
    gdbgi("  opt: http-ip=[%s]", config->http_ip);
    gdbgi("  opt: http-port=[%d]", config->http_port);
    gdbgi("  opt: tlv-path=[%s]", config->tlv_path);
    gdbgi("  opt: job-path=[%s]", config->job_path);
}


int main (int argc, char **argv){
    
    xm_crash_init(argc, argv);
    
    if(app_config_parse(argc, argv)){
        return 1;
    }
    
    app_config * config = app_config_get();
    if(config->http_port == 0){
        config->http_port = 1770;
    }
    
    xm_log_init(config->log_config_file_name);
    
    gdbgi("APP_VER = %s", APP_VER_STR);
    dump_config(config);
    if(config->media_ip && config->media_ip[0] != '\0'){
        if(check_localip(config->media_ip)){
            gdbge("fail to discovery media_ip %s", config->media_ip);
            return 1;
        }
        gdbgi("successfully check media_ip %s", config->media_ip);
    }
    
    const char * cert_file = NULL; // "/Users/simon/Desktop/simon/projects/easemob/src/xmedia/dtls/certs/server-cert.pem";
    const char * key_file = NULL; // "/Users/simon/Desktop/simon/projects/easemob/src/xmedia/dtls/certs/server-key.pem";
    int ret = xdtls_srtp_init(cert_file, key_file);
    if(ret < 0){
        gdbge("xdtls_srtp_init fail with %d", ret);
        return 1;
    }
    
//    websocket_client_main(argc, argv);
    websocket_server_main(argc, argv);
    
    return 0;
}
